package pku;

/**
 * Created by yangxiao on 2017/11/14.
 */
public class Producer {
    public ByteMessage createBytesMessageToTopic(String topic, byte[] body)throws Exception{
    	ByteMessage msg=new DefaultMessage(body);
        msg.putHeaders(MessageHeader.TOPIC,topic);
        return msg;
    }
    public void send(ByteMessage defaultMessage)throws Exception{
    	 String topic = defaultMessage.headers().getString(MessageHeader.TOPIC);
         MessageStore.store.push(defaultMessage,topic);
    }
    public void flush()throws Exception{ 
    	MessageStore.store.writeObjectToFile(MessageStore.store.msgs,"0");
    	while(true){
    		if(MessageStore.store.threadMap.get(Thread.currentThread().getId()).isEmpty()){
    			break;
    		} else {
    			Thread.sleep(300);
    		}
    	}
    	System.out.println(1);
    }
}
